package org.geektime.reactive.message.streams.v1;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

import java.util.LinkedList;
import java.util.List;

/**
 * @author lanxing
 */
public class DefaultPublisher<T> implements Publisher<T> {

    private List<SubscriberWrapper> subscribers = new LinkedList<>();

    @Override
    public void subscribe(Subscriber<? super T> s) {
        DefaultSubscription subscription = new DefaultSubscription(s);
        s.onSubscribe(subscription);
        subscribers.add(new SubscriberWrapper(s,subscription));
    }

    public void publish(T data){
        subscribers.forEach(subscriber -> {
            SubscriberWrapper subscriberWrapper = SubscriberWrapper.class.cast(subscriber);
            DefaultSubscription defaultSubscription = subscriberWrapper.getDefaultSubscription();
            if(defaultSubscription.isCanceled()){
                System.out.println("本次数据发布已忽略:"+ data);
                return;
            }
            //继续发送
            subscriber.onNext(data);
        });
    }

    public void error(Throwable error){
        subscribers.forEach(subscriber -> {
            subscriber.onError(error);
        });
    }

    public void complate(){
        // 广播
        subscribers.forEach(subscriber -> {
            subscriber.onComplete();
        });
    }


    public static void main(String[] args) {
        DefaultPublisher defaultPublisher = new DefaultPublisher();
        defaultPublisher.subscribe(new DefaultSubscriber());
        for(int i= 0 ;i< 5; i++){
            defaultPublisher.publish(i);
        }
    }
}
